﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Net.Mime;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using NetMQ;
using NetMQ.Sockets;

namespace Lawn.AppServer
{
    
    
    public class ServerMq:IServerMq
    {


        private ILogger<ServerMq> logger;
        private NetMQPoller _poller;
        // private SubscriberSocket _subscriberSocket;
        private ISubHandler  _subHandler;
         private PublisherSocket _publisherSocket;
        private ITopicManager _topicManager;

        private OptionsPort optionsPort;

        private List<OptionsRemoteServer>  optionsRemoteServer;

        private const int HEAT_TIME_SPAN = 1000;

        private NetMQTimer heatNetMqTimer;

        private IConfiguration configuration;
        public ServerMq(ILogger<ServerMq> logger,
            ISubHandler subHandler, 
            ITopicManager topicManager,
            IOptions<OptionsPort> _optionsPort,
            IOptions<List<OptionsRemoteServer>> _optionsRemoteServer,
            IConfiguration configuration)
        {

            this.configuration = configuration;
            this.logger = logger;
            this.optionsRemoteServer = _optionsRemoteServer.Value;
            this.optionsPort = _optionsPort.Value; 
            this._poller = new NetMQPoller();
            this._subHandler = subHandler;
            this.heatNetMqTimer = new NetMQTimer(TimeSpan.FromMilliseconds(HEAT_TIME_SPAN));
            this._publisherSocket = new PublisherSocket(); 
            this._topicManager = topicManager;
            this.heatNetMqTimer.Elapsed += PubHeatMessage;
            this._poller.Add(heatNetMqTimer);
             

        }

        private void PubHeatMessage(object o, NetMQTimerEventArgs e)
        {
            if (_publisherSocket != null)
            {
                var id = DateTime.Now.Ticks;
                var msg = BitConverter.GetBytes(id);
                _publisherSocket.SendMoreFrame("HEAT").TrySendFrame(msg);
                logger.LogDebug($"send heat:{id}");
            }
        }
        


        private ConcurrentDictionary<string,NetMQSocket> subSockets=new ConcurrentDictionary<string,NetMQSocket>();
        private void DynamicsAddSubSocket()
        {

            foreach (var  server in optionsRemoteServer)
            {
                var sub = new SubscriberSocket();
                sub.Connect($"tcp://{server.Url}:{server.Port}");
                logger.LogInformation($"start connect:{server.Url}:{server.Port}");
                foreach (var topic in server.TopicTypes)
                {
                    sub.Subscribe(topic.Topic);
                    sub.ReceiveReady += (o, e) =>
                    {
                        _subHandler.Handler(e.Socket,_publisherSocket,server.Key);
                    };
                }

                subSockets.TryAdd(server.Key, sub);
                _poller.Add(sub);
                
            }

            
            
        }

        public  Task Start()
        {
            
           
        //    logger.LogInformation(configuration["Server:Port:PubPort"]);
            
          //  logger.LogInformation($"----------{_environment.EnvironmentName}");
            
            DynamicsAddSubSocket();
            logger.LogInformation($"server port:{optionsPort.PubPort}");  
            _publisherSocket.Bind($"tcp://*:{optionsPort.PubPort}");  
            _poller.RunAsync();
            
            return Task.CompletedTask;
        }


        private async Task DynamicsRemoveSubSocket()
        {
            //取消订阅，

            foreach (var key in subSockets.Keys)
            {

                NetMQSocket socket = null;
                subSockets.TryGetValue(key, out socket);
                if (socket!=null)
                {
                    var topics = await _topicManager.GetTopicByKey(key); 
                    foreach (var topic in topics)
                    {
                        (socket as SubscriberSocket)?.Unsubscribe(topic.Topic);
                    } 
                    _poller.RemoveAndDispose(socket);
                }
                
            }
            
        }
        public    Task Stop()
        {
            
            logger.LogInformation("clear.........");
            DynamicsRemoveSubSocket().ConfigureAwait(false);
            heatNetMqTimer.Enable = false;
            _poller.Remove(heatNetMqTimer);
            _poller.StopAsync();
           
            _poller.Dispose(); 
            _publisherSocket.Dispose();
            NetMQConfig.Cleanup();
            
            logger.LogInformation("cleared.........");
            return  Task.CompletedTask;
            
        }
    }
}